package cn.tianyigps.tianyigpsm7clean.handler;

import cn.tianyigps.tianyigpskafka.task.bean.Task;
import cn.tianyigps.tianyigpskafka.task.manger.TaskManger;
import cn.tianyigps.tianyigpsm7clean.assign.EquipmentAssign;
import cn.tianyigps.tianyigpsm7clean.dao.EquipmentDao;
import cn.tianyigps.tianyigpspublic.bean.EquipmentModel;
import cn.tianyigps.tianyigpspublic.bean.StandardModel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.stream.Collectors;


/**
 * 清洗定时任务
 * Created with IDEA
 * author:guibin.cui
 * Date:2018/5/25
 * Time:9:22
 */
@Component
@Slf4j
public class CleanHandler {

    @Autowired
    private Executor executor;
    @Autowired
    private EquipmentDao equipmentDao;
    @Autowired
    private EquipmentAssign equipmentAssign;
    @Autowired
    private TaskManger taskManger;


    /**
     * 设备信息处理
     */
    public void handlerEquipment(List<StandardModel> standardModels) {
        // 根据设备id分组
        Map<String, List<StandardModel>> collect = standardModels.stream()
                .filter(standardModel -> !StringUtils.isEmpty(standardModel.getEquipmentId()))
                .collect(Collectors.groupingBy(StandardModel::getEquipmentId));

        CountDownLatch countDownLatch = new CountDownLatch(collect.size());
        List<EquipmentModel> equipmentModels = new CopyOnWriteArrayList<>();
        log.info("主线程：name" + Thread.currentThread().getName());

        collect.forEach((equipmentId, standardModelList) -> {
            log.info("equipmentId: " + equipmentId + "  standardModelList.size : " + standardModelList.size());
            executor.execute(() -> {
                try {
                    doClean(equipmentModels,standardModelList);
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("设备：equipmentId " + equipmentId + ",   清洗失败 :" + standardModelList);
                } finally {
                    countDownLatch.countDown();
                }
            });
        });
        try {
            //主线程等待子线程集合执行完成
            log.info("主线程等待：name:  " + Thread.currentThread().getName());
            countDownLatch.await();
            log.info("主线程释放：name:  " + Thread.currentThread().getName());

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 保存轨迹
        savaLocus(equipmentModels);
    }
    private void doClean(List<EquipmentModel> equipmentModels, List<StandardModel> standardModelList) {

        if (!Optional.ofNullable(standardModelList).isPresent() || standardModelList.isEmpty())
            return;
        List<StandardModel> standardModels = standardModelList.stream().sorted(Comparator.comparing(StandardModel::getReportTime)).collect(Collectors.toList());
        if (!Optional.ofNullable(standardModels).isPresent() || standardModels.isEmpty())
            return;

        EquipmentModel equipmentModel = equipmentDao.findByEquipmentId(standardModels.get(0).getEquipmentId());
        if (!Optional.ofNullable(equipmentModel).isPresent()) {
            equipmentModel = new EquipmentModel();
            equipmentModel.setActiveTime(new Date());
        }
        // 补全的原始数据
        List<StandardModel> sptStandardModels = standardModels.stream().filter(standardModel -> "01".equals(standardModel.getRtr())).collect(Collectors.toList());
        EquipmentModel pEquipmentModel = new EquipmentModel();
        BeanUtils.copyProperties(equipmentModel, pEquipmentModel);
        standardModels.removeAll(sptStandardModels);
        // 实时数据
        equipmentAssign.assgin(equipmentModel, standardModels);
        equipmentDao.save(equipmentModel);
        sendTask(equipmentModel);
        equipmentModel.setId(null);
        Date newLocateTime = equipmentModel.getLocateTime();
        Date oldLocateTime = pEquipmentModel.getLocateTime();
        if (!Optional.ofNullable(oldLocateTime).isPresent())
            oldLocateTime = new Date();
        if(Optional.ofNullable(newLocateTime).isPresent() && !newLocateTime.equals(oldLocateTime)){
            equipmentModels.add(equipmentModel);
        }

        // 补全数据插入数据库
        equipmentAssign.assginSupplement(equipmentModels, sptStandardModels);
    }

    private void savaLocus(List<EquipmentModel> equipmentModels) {

        Date date = new Date();
        DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
        String format = dateFormat.format(date);
        equipmentDao.insertBatch(equipmentModels,"m7Equipment-" + format);

    }
    private void sendTask(EquipmentModel equipmentModel) {
        Task task = new Task();
        task.setTaskType("m7_clean");
        Map<String, Object> params = new HashMap<>();
        params.put("equipmentModel",equipmentModel);
        task.setParams(params);
        taskManger.distributeTask(task);
    }

}
